package com.shujia.wyh.base;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.List;

/*
    专用过滤器&包装过滤器
 */
public class HbaseFilterDemo2 {
    private Connection conn;
    private Admin admin;

    @Before
    public void init() {
        try {
            //获取与Hbase的连接对象
            //1、获取Hadoop的相关环境
            Configuration conf = new Configuration();

            //2、设置以下zookeeper的集群信息
            conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

            //3、获取habse的连接对象
            conn = ConnectionFactory.createConnection(conf);

            //4、获取HMaster的对象
            admin = conn.getAdmin();

            System.out.println("获取连接成功！！" + conn);
            System.out.println("HMaster的对象获取成功！！" + admin);
            System.out.println("------------------------------------------------------------------");


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 通过SingleColumnValueFilter与查询文科班所有学生信息
     */
    @Test
    public void singleColumnValueFun() {
        try {
            //获取表的实例
            //先将表封装成TableName的对象
            TableName students2 = TableName.valueOf("students2");

            //获取表的实例对象
            Table table = conn.getTable(students2);

            Scan scan = new Scan();

            //创建二进制前缀比较器
            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator(Bytes.toBytes("文科"));

            //创建单列值过滤器
            //ingleColumnValueFilter(final byte [] family, final byte [] qualifier,final CompareOperator op,final org.apache.hadoop.hbase.filter.ByteArrayComparable comparator)
            SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                    Bytes.toBytes("shujia"),
                    Bytes.toBytes("班级"),
                    CompareOperator.EQUAL,
                    binaryPrefixComparator);

            scan.setFilter(singleColumnValueFilter);

            //查询数据
            ResultScanner resultScanner = table.getScanner(scan);

            printResultScanner(resultScanner);


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * SingleColumnValueExcludeFilter列值排除过滤器
     */
    @Test
    public void singleColumnValueExcludeFilterFun() {
        try {
            //获取表的实例
            //先将表封装成TableName的对象
            TableName students2 = TableName.valueOf("students2");

            //获取表的实例对象
            Table table = conn.getTable(students2);

            Scan scan = new Scan();

            //创建二进制前缀比较器
            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator(Bytes.toBytes("文科"));

            //创建列值排除过滤器
            //public SingleColumnValueExcludeFilter(byte[] family, byte[] qualifier,CompareOperator op, ByteArrayComparable comparator)
            SingleColumnValueExcludeFilter singleColumnValueExcludeFilter = new SingleColumnValueExcludeFilter(
                    Bytes.toBytes("shujia"),
                    Bytes.toBytes("班级"),
                    CompareOperator.EQUAL,
                    binaryPrefixComparator);


            scan.setFilter(singleColumnValueExcludeFilter);

            //查询数据
            ResultScanner resultScanner = table.getScanner(scan);

            printResultScanner(resultScanner);


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 使用PrefixFilter前缀过滤器查询以15001001开头的所有前缀的rowkey
     */
    @Test
    public void prefixFilterFun() {
        try {
            //获取表的实例
            //先将表封装成TableName的对象
            TableName students2 = TableName.valueOf("students2");

            //获取表的实例对象
            Table table = conn.getTable(students2);

            Scan scan = new Scan();

            //创建行键前缀过滤器
            PrefixFilter prefixFilter = new PrefixFilter(Bytes.toBytes("15001001"));


            scan.setFilter(prefixFilter);

            //查询数据
            ResultScanner resultScanner = table.getScanner(scan);

            printResultScanner(resultScanner);


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 分页过滤器 需求：前30行分3页出来
     */
    @Test
    public void pageFilterFun() {
        try {
            //将表名封装成TableName的对象
            TableName students2 = TableName.valueOf("students2");

            //获取表的实例对象
            Table table = conn.getTable(students2);

            //定义你要查询的页数
            int pageNum = 3;

            //定义每一页的行数
            int pageSize = 10;

            //定义开始行数的变量
            String currentPageStartRow = "";

            //创建一个Scan对象
            Scan scan = new Scan();

            for (int i = 1; i <= pageNum; i++) {
                System.out.println("================================当前是第"+i+"页=====================================");
                //创建分页过滤器
                PageFilter pageFilter = new PageFilter(pageSize);
                scan.setFilter(pageFilter);
                ResultScanner resultScanner = table.getScanner(scan);
//                printResultScanner(resultScanner);
                //获取到每一行
                for (Result result : resultScanner) {
                    //获取当前行的rowkey+0作为下一页的开始行
                    currentPageStartRow = Bytes.toString(result.getRow())+0;
                    scan.withStartRow(Bytes.toBytes(currentPageStartRow));
                    //学号
                    String id = Bytes.toString(result.getRow());
                    String name = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("姓名")));
                    String age = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("年龄")));
                    String gender = Bytes.toString(result.getValue(Bytes.toBytes("shujia"), Bytes.toBytes("性别")));
                    String clazz = Bytes.toString(result.getValue(Bytes.toBytes("shujia"), Bytes.toBytes("班级")));
                    System.out.println("学号："+id+", 姓名："+name+", 年龄："+age+", 性别："+gender+", 班级"+clazz);
                }
            }


        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    @After
    public void close() {
        if (admin != null) {
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 自定义一个方法输出getScanner的结果值
     */
    public void printResultScanner(ResultScanner resultScanner) {
        try {
            //定义一个变量是Result类型的
            Result rs = null;
            //将每一次获取的值赋值给rs,如果rs不为null,说明获取到值
            //rs获取的是一行的值，而一行可以有若干个列
            while ((rs = resultScanner.next()) != null) {
                //获取到行键值
                String id = Bytes.toString(rs.getRow());
                //上面的获取，是在我们已经知道了列名的前提下获取的，如果我们不知道列名，又怎么获取呢？
                //获取一行中所有的列，每一列组成了Cell对象，返回一个集合
                List<Cell> cells = rs.listCells();

                StringBuilder sb = new StringBuilder();
                sb.append("学号：").append(id);

                for (Cell cell : cells) {
                    String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    sb.append(", ").append(qualifier).append("：").append(value);
                }

                //得到一行的拼接结果
                System.out.println(sb);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
